package org.example;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class WordCount {
    public static void main(String[] args) {
        System.out.println("Hello world!");
        // 接收参数
        String srcFilePath = args[0];
        String outFilePath = args[1];
        System.out.println("输入路径:"+srcFilePath);
        System.out.println("输出路径:"+outFilePath);
        // 配置环境
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("WordCount");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 读取文件数据
        JavaRDD<String> stringRDD = sc.textFile(srcFilePath);
        JavaRDD<String> wordStringRDD = stringRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });
        // 转换成key-value形式
        JavaPairRDD<String,Integer> wordStringCountRDD = wordStringRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word,1);
            }
        });
        // 进行汇聚
        JavaPairRDD<String,Integer> wordStringSumRDD = wordStringCountRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer1, Integer integer2) throws Exception {
                return integer1+integer2;
            }
        });
        // 进行输出
        wordStringSumRDD.saveAsTextFile(outFilePath);
        sc.close();
    }
}